Skip to content

Fix bsr watcher leak - #873

Merged
jem-davies merged 7 commits into
warpstreamlabs:mainfrom
jem-davies:fix-bsr-watcher-leak
Jun 4, 2026
Merged

Fix bsr watcher leak#873
jem-davies merged 7 commits into
warpstreamlabs:mainfrom
jem-davies:fix-bsr-watcher-leak

Conversation

@jem-davies

@jem-davies jem-davies commented Jun 1, 2026

Copy link
Copy Markdown
Collaborator

So the "buf schema registry" poller appears to be leaking because it's context is not cancelled on Processor Close().

This PR adds 3 commits:

  1. Adds goleak to detect goroutine leaks in the ./internal/impl/protobuf tests
  2. Adds context cancellation to the schema watcher
  3. Removes an fmt.Print call
  4. fixes the linter

...

Previously I have been implementing the "lease" & "cache" option for this processor on #870 - however I think the issue of frequent calls to BSR services will likely be fixed with this PR.

Signed-off-by: Jem Davies <jemsot@gmail.com>
Signed-off-by: Jem Davies <jemsot@gmail.com>
@jem-davies
jem-davies requested a review from gregfurman as a code owner June 1, 2026 11:57
Comment thread internal/impl/protobuf/processor_protobuf.go Outdated
Signed-off-by: Jem Davies <jemsot@gmail.com>
Comment thread internal/impl/protobuf/multimodule_watcher.go
}
multiModuleWatcher := &MultiModuleWatcher{}

ctx, cancel := context.WithCancel(context.Background())

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We shouldn't really be persisting ctx types. I think this should rather be a shutdown.Signaller that creates a new ctx.

e.g.

ctx, cancel := multiModuleWatcher.shutSig.SoftStopCtx(context.Background())


func (m *MultiModuleWatcher) Close() {
m.cancel()
for _, v := range m.bsrClients {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Q: Is it a memory leak that we never actually clear these clients stored in the map? Also, what if one stops before we call close -- should it be deleted early? 🤔

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added a m.bsrClients = nil - It should be ok to call Stop() multiple times.

func (p *protobufProc) Close(context.Context) error {
func (p *protobufProc) Close(ctx context.Context) error {
if p.multiModuleWatcher != nil {
p.multiModuleWatcher.Close()

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we be setting p.multiModuleWatcher = nil here?

Signed-off-by: Jem Davies <jemsot@gmail.com>
Signed-off-by: Jem Davies <jemsot@gmail.com>
@jem-davies

Copy link
Copy Markdown
Collaborator Author

@gregfurman - thanks for review - I have responded / added commits to the code 🙏

Signed-off-by: Jem Davies <jemsot@gmail.com>
@jem-davies
jem-davies merged commit 5f78b91 into warpstreamlabs:main Jun 4, 2026
4 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants